package com.sl.cdc.modular.debezium;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.*;
import com.google.common.collect.Maps;
import com.sl.cdc.api.CdcConfigApi;
import com.sl.cdc.api.domain.*;
import com.sl.cdc.api.enums.CdcDataSourceEnum;
import com.sl.cdc.api.enums.SnapshotMode;
import io.debezium.embedded.EmbeddedEngine;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;


@Slf4j
public class CdcDebeziumManager implements CdcConfigApi {

    private ExecutorService executor = Executors.newSingleThreadExecutor();

    private final Map<String, DebeziumEngine<ChangeEvent<String, String>>> debeziumEngineMap = Maps.newConcurrentMap();


    @Override
    public String addTask(TaskInfo taskInfo) {

        Long snowflakeNextId = IdUtil.getSnowflakeNextId();
        String id = taskInfo.getId();

        if (StrUtil.isBlank(id)) {
            id = snowflakeNextId.toString();
        }
        //先停止
        close(id);

        String property = System.getProperty("user.home");

        Properties properties = taskInfo.getProperties();

        TaskDataSourceInfo sourceInfo = taskInfo.getSourceInfo();
        CdcDataSourceEnum sourceDs = sourceInfo.getSourceDs();
        // Define the configuration for the Debezium Engine with MySQL connector...
        final Properties props = new Properties();
        props.setProperty("name", StrUtil.nullToDefault(taskInfo.getName(), id));
        props.setProperty("connector.class", sourceDs.getConnector());
        props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
        String path  =property + "/square_lake/tmp/offsets/" + id + ".dat";
        createFile(path);
        props.setProperty("offset.storage.file.filename", path);
        //尝试提交偏移量的时间间隔。默认值为1分钟。
        props.setProperty("offset.flush.interval.ms", "60000");
        //设置时区
        props.setProperty("database.serverTimezone", "Asia/Shanghai");

        props.put("converters", "timestampConverter");
        props.put("timestampConverter.type", "com.sl.cdc.core.config.TimestampConverter");
        props.put("timestampConverter.format.time", "HH:mm:ss");
        props.put("timestampConverter.format.date", "YYYY-MM-dd");
        props.put("timestampConverter.format.datetime", "YYYY-MM-dd HH:mm:ss");
        props.put("timestampConverter.format.timestamp", "YYYY-MM-dd HH:mm:ss.SSS");


        //include.schema.changes
        /* begin connector properties */
        props.setProperty("database.hostname", sourceInfo.getHostname());
        props.setProperty("database.port", sourceInfo.getPort());
        props.setProperty("database.user", sourceInfo.getUser());
        props.setProperty("database.password", sourceInfo.getPassword());
        //initial
        props.setProperty("snapshot.mode", SnapshotMode.SCHEMA_ONLY.getCode());
        props.setProperty("database.server.id", id);
        props.setProperty("database.server.name", props.getProperty("name") + "Server");
        props.setProperty("topic.prefix", "square-lake-" + props.getProperty("name") + "-connector");
        props.setProperty("schema.history.internal", "io.debezium.storage.file.history.FileSchemaHistory");
        props.put("decimal.handling.mode", "string");
        props.put("database.tinyInt1isBit", "false");


        String s2 = buildDatabaseIncludeList(sourceInfo);
        if (StrUtil.isNotBlank(s2)) {
            props.setProperty("database.include.list", s2);
        }
        String s1 = buildTableIncludeList(sourceInfo);
        if (StrUtil.isNotBlank(s1)) {
            props.setProperty("table.include.list", s1);
        }

        String s = buildColumnIncludeList(sourceInfo);
        if (StrUtil.isNotBlank(s)) {
            props.setProperty("column.include.list", s);
        }

        //schema的变更不管，现在就是根据配置的信息写表
        props.setProperty("include.schema.changes", "false");

        props.setProperty("schema.history.internal.file.filename", property + "/square_lake/tmp/history/" + id + ".dat");


        if (ObjUtil.isNotNull(properties)) {
            props.putAll(properties);
        }
        CdcDebeziumConsumer cdcDebeziumConsumer = new CdcDebeziumConsumer(taskInfo);
        DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class).using(props).notifying(cdcDebeziumConsumer).using(new CdcDebeziumCompletionCallback(id)).using(new CdcDebeziumConnectorCallback()).build();
        debeziumEngineMap.put(id, engine);
        return id;
    }
    @SneakyThrows
    private void createFile(String path ){

        File file = FileUtil.file(path);
        if(!file.exists()){
            FileUtil.touch(file);
        }
    }

    private String buildColumnIncludeList(TaskDataSourceInfo sourceInfo) {
        if (ObjectUtil.isNull(sourceInfo)) {
            return null;
        }
        List<TaskDataBaseInfo> dataBaseList = sourceInfo.getDataBaseList();
        if (CollUtil.isEmpty(dataBaseList)) {
            return null;
        }
        String collect = dataBaseList.stream().filter(v -> {
            return CollUtil.isNotEmpty(v.getTableList());
        }).map(v -> {
            String code = v.getCode();
            List<TaskTableInfo> tableList = v.getTableList();
            return tableList.stream().map(t -> {
                String table = t.getCode();
                List<TaskColumnInfo> columnList = t.getColumnList();
                return columnList.stream().map(vv -> {
                    return StrUtil.concat(true, code, ".", table, ".", vv.getCode());
                }).collect(Collectors.toSet());
            }).flatMap(Collection::stream).collect(Collectors.toSet());
        }).flatMap(Collection::stream).collect(Collectors.joining(","));
        return collect;
    }

    private String buildTableIncludeList(TaskDataSourceInfo sourceInfo) {
        if (ObjectUtil.isNull(sourceInfo)) {
            return null;
        }
        List<TaskDataBaseInfo> dataBaseList = sourceInfo.getDataBaseList();

        String collect = dataBaseList.stream().filter(v -> {
            return CollUtil.isNotEmpty(v.getTableList());
        }).map(v -> {
            String code = v.getCode();
            List<TaskTableInfo> tableList = v.getTableList();

            return tableList.stream().map(t -> StrUtil.concat(true, code, "." + t.getCode())).collect(Collectors.toSet());
        }).flatMap(Collection::stream).collect(Collectors.joining(","));
        return collect;
    }


    private String buildDatabaseIncludeList(TaskDataSourceInfo sourceInfo) {
        if (ObjectUtil.isNull(sourceInfo)) {
            return null;
        }
        List<TaskDataBaseInfo> dataBaseList = sourceInfo.getDataBaseList();
        return dataBaseList.stream().map(TaskDataBaseInfo::getCode).collect(Collectors.joining(","));
    }


    @Override
    public Boolean start() {
        return this.start(null);
    }

    @Override
    public String getStatus(String id) {
        String status = "STOP";
        for (Map.Entry<String, DebeziumEngine<ChangeEvent<String, String>>> debeziumEngine : debeziumEngineMap.entrySet()) {
            DebeziumEngine value = debeziumEngine.getValue();
            String key = debeziumEngine.getKey();
            if (StrUtil.isNotBlank(id)) {
                if (!StrUtil.equals(key, id)) {
                    continue;
                }
            }
            if (value instanceof EmbeddedEngine engine) {
                boolean running = engine.isRunning();
                status = running ? "RUN" : "STOP";
                break;
            }
        }
        return status;
    }

    @Override
    public Boolean start(String id) {
        for (Map.Entry<String, DebeziumEngine<ChangeEvent<String, String>>> debeziumEngine : debeziumEngineMap.entrySet()) {
            DebeziumEngine value = debeziumEngine.getValue();
            String key = debeziumEngine.getKey();
            if (StrUtil.isNotBlank(id)) {
                if (!StrUtil.equals(key, id)) {
                    continue;
                }
            }
            Object obj = value;
            Object cb = ReflectUtil.getFieldValue(value, "val$engine");
            if (ObjectUtil.isNotNull(cb)) {
                obj = cb;
            }

            if (obj instanceof EmbeddedEngine engine) {
                boolean running = engine.isRunning();
                if (!running) {
                    executor.execute(value);
                }
            }
        }
        return Boolean.TRUE;
    }

    @SneakyThrows
    @Override
    public Boolean close() {
        return this.close(null);
    }

    @SneakyThrows
    @Override
    public Boolean close(String id) {
        for (Map.Entry<String, DebeziumEngine<ChangeEvent<String, String>>> debeziumEngine : debeziumEngineMap.entrySet()) {
            DebeziumEngine value = debeziumEngine.getValue();
            String key = debeziumEngine.getKey();
            if (StrUtil.isNotBlank(id)) {
                if (!StrUtil.equals(key, id)) {
                    continue;
                }
            }
            value.close();
        }
        return Boolean.TRUE;
    }

    @Override
    public Boolean remove(String id) {
        this.close(id);
        debeziumEngineMap.remove(id);
        return Boolean.TRUE;
    }


    private String buildStorageName(TaskInfo taskInfo) {
        TaskDataSourceInfo sourceInfo = taskInfo.getSourceInfo();
        TaskDataSourceInfo targetInfo = taskInfo.getTargetInfo();

        String name = sourceInfo.getName();
        String hostname = sourceInfo.getHostname();
        if (ObjUtil.isNull(targetInfo)) {
            targetInfo = new TaskDataSourceInfo();

        }
        String targetInfoName = targetInfo.getName();
        String targetInfoHostname = targetInfo.getHostname();

        return StrUtil.concat(true, name, "_", hostname, "-to-", targetInfoName, "_", targetInfoHostname);
    }

}
